package cn.z2huo.demo.hello.kafka.impl;

import cn.z2huo.demo.hello.core.common.TransactionCodeHolder;
import cn.z2huo.demo.hello.core.common.util.PropertiesUtils;
import cn.z2huo.demo.hello.core.common.util.TransactionUtils;
import cn.z2huo.demo.hello.kafka.api.*;
import cn.z2huo.demo.hello.kafka.support.ConsumerRunner;
import cn.z2huo.demo.hello.kafka.util.KafkaConfig;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.schedulers.Schedulers;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.context.ApplicationContext;

import java.util.concurrent.Future;

/**
 * <p>
 *
 * @author z2huo
 */
@Slf4j
@RequiredArgsConstructor
public class KafkaLauncher implements KafkaOperations {

    private final KafkaConnectionFactory kafkaConnectionFactory;

    private final KafkaConsumerRunnerDAO kafkaConsumerRunnerDAO;

    private final ApplicationContext applicationContext;

    @Override
    public void send(String producerName, ProducerRecord<String, String> record, Callback callback) {
        KafkaProducer<String, String> producer = kafkaConnectionFactory.getProducer(producerName);
        producer.send(record, callback);
    }

    @Override
    public Future<RecordMetadata> send(String producerName, ProducerRecord<String, String> record) {
        KafkaProducer<String, String> producer = kafkaConnectionFactory.getProducer(producerName);
        return producer.send(record);
    }

    @Override
    public void subscribe(String consumerName, KafkaSubscriber subscriber) {

        String groupId = PropertiesUtils.resolvePlaceholders("${" + KafkaConfig.CONSUMER_PROP_PREFIX + consumerName
                        + ".group.id:${" + KafkaConfig.PRODUCER_PROP_LEGACY_PREFIX + consumerName + ".group.id:}}");

        if (StringUtils.isNotBlank(subscriber.groupId())) {
            groupId = subscriber.groupId();
        }

        String runnerName = KafkaConfig.generateRunnerName(consumerName, groupId, subscriber.topics());

        Observable.just(kafkaConnectionFactory).observeOn(Schedulers.io()).subscribe(factory -> {

            TransactionCodeHolder.set(TransactionUtils.generateTransactionCode());

            KafkaConsumable consumerRunner = null;
            try {
                if (subscriber.parallel()) {
                    // TODO 2024/8/30 这里要替换成多线程消费 runner
                    consumerRunner = new ConsumerRunner(consumerName, subscriber, applicationContext, factory);
                } else {
                    consumerRunner = new ConsumerRunner(consumerName, subscriber, applicationContext, factory);
                }
            } catch (Exception e) {
                log.error("occur an error with topics = [" + String.join(",", subscriber.topics()) + "]", e);
                return;
            }

            kafkaConsumerRunnerDAO.add(runnerName, consumerRunner);

            consumerRunner.run();
        });

    }
}
